package com.katesoft.scale4j.rttp.tools;

import static com.katesoft.scale4j.common.lang.RuntimeUtility.VAR_HAZELCAST_CONFIGURATION;

import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import net.jcip.annotations.ThreadSafe;

import org.apache.commons.lang.builder.ToStringBuilder;
import org.springframework.beans.BeanInstantiationException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.Resource;

import com.hazelcast.config.Config;
import com.hazelcast.config.UrlXmlConfig;
import com.hazelcast.config.XmlConfigBuilder;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.hibernate.IHazelcastInstanceLoader;
import com.hazelcast.impl.GroupProperties;
import com.katesoft.scale4j.log.LogFactory;
import com.katesoft.scale4j.log.Logger;

/**
 * this class is responsible for starting hazelcast instance during spring context initialization.
 * Clients may want to specify location of hazelcast.xml configuration file(it is possible to do
 * this using -Dapp.hazelcast.config property), but this is not necessary.
 * <p/>
 * It is also possible to inject configuration location into this class directly.
 * 
 * @author kate2007
 */
@ThreadSafe
public class HazelcastLauncher implements IHazelcastInstanceLoader, DisposableBean,
         InitializingBean {
   private static final Logger LOGGER = LogFactory.getLogger(HazelcastLauncher.class);
   //
   private Resource configLocation;
   private HazelcastInstance hazelcastInstance;
   private final Lock initializationLock = new ReentrantLock();

   @Override
   public HazelcastInstance loadInstance(@SuppressWarnings("unused") Properties props) {
      initializationLock.lock();
      try {
         if (hazelcastInstance == null) {
            init();
         }
      } finally {
         initializationLock.unlock();
      }
      return hazelcastInstance;
   }

   private void init() {
      if (hazelcastInstance == null) {
         long time = System.currentTimeMillis();
         Config config;
         if (System.getProperty(VAR_HAZELCAST_CONFIGURATION) != null) {
            LOGGER.info("launching hazelcast instance from config location %s",
                     System.getProperty(VAR_HAZELCAST_CONFIGURATION));
            try {
               config = new UrlXmlConfig(System.getProperty(VAR_HAZELCAST_CONFIGURATION));
            } catch (IOException e) {
               LOGGER.error(e);
               throw new BeanInstantiationException(HazelcastLauncher.class, String.format(
                        "unable to init hazelcast from location %s",
                        System.getProperty(VAR_HAZELCAST_CONFIGURATION)), e);
            }
         } else if (configLocation == null) {
            LOGGER.info("launching hazelcast instance from 'hazelcast-default.xml'");
            config = new XmlConfigBuilder().build();
         } else {
            try {
               config = new UrlXmlConfig(configLocation.getURL());
               LOGGER.info("launching hazelcast instance from config location %s", configLocation
                        .getURL().toString());
            } catch (IOException e) {
               LOGGER.error(e);
               throw new BeanInstantiationException(HazelcastLauncher.class, String.format(
                        "unable to init hazelcast from location %s",
                        configLocation.getDescription()), e);
            }
         }
         if (!config.getProperties().contains(GroupProperties.PROP_WAIT_SECONDS_BEFORE_JOIN)) {
            config.getProperties().put(GroupProperties.PROP_WAIT_SECONDS_BEFORE_JOIN, "1");
         }
         hazelcastInstance = Hazelcast.newHazelcastInstance(config);
         LOGGER.info("Hazelcast instance launched in [%s] milisecs and listening port = %s",
                  (System.currentTimeMillis() - time), config.getPort());
         Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
               LOGGER.info("hazelcast shutdown hook triggered");
               stop();
            }
         }));
      }
   }

   /**
    * setter for hazelcast configuration file location.
    * 
    * @param configLocation
    *           URL
    */
   public void setConfigLocation(Resource configLocation) {
      this.configLocation = configLocation;
   }

   public void stop() {
      initializationLock.lock();
      try {
         if (hazelcastInstance != null) {
            long time = System.currentTimeMillis();
            if (hazelcastInstance.getLifecycleService().isRunning()) {
               String address = hazelcastInstance.getCluster().getLocalMember().toString();
               LOGGER.info("stopping hazelcast instance[%s]", address);
               hazelcastInstance.getLifecycleService().shutdown();
               LOGGER.info("Hazelcast instance [%s] stopped in [%s] milisecs", address,
                        (System.currentTimeMillis() - time));
            }
         }
         hazelcastInstance = null;
      } finally {
         initializationLock.unlock();
      }
   }

   public HazelcastInstance getHazelcastInstance() {
      return hazelcastInstance;
   }

   public boolean startedFromSystemParam() {
      return System.getProperty(VAR_HAZELCAST_CONFIGURATION) != null;
   }

   @Override
   public String toString() {
      return new ToStringBuilder(this).append("configLocation", configLocation)
               .append("-Dapp.configuration", System.getProperty(VAR_HAZELCAST_CONFIGURATION))
               .toString();
   }

   @Override
   public void destroy() {
      stop();
   }

   @Override
   public void afterPropertiesSet() {
      init();
   }
}
